#include "Logger.h"
#include "EventLoop.h"
#include "Poller.h"
#include "Channel.h"


#include <sys/eventfd.h>
#include <unistd.h>
#include <fcntl.h>
#include <cerrno>

// 防止一个线程创建多个EventLoop 
__thread EventLoop* t_loopInThisThread = nullptr;

// 定义默认的IO复用接口的最大超时时间
const int kPollTimeMs = 10000;

// 创建wakeupfd用来通知/唤醒subReactor处理新来的Channel
int createEventFd()
{
    int evtfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if (evtfd < 0) {
        LOG_FATAL("eventfd error:%d \n", errno);
    }

    return evtfd;
}

EventLoop::EventLoop() : m_looping(false), m_quit(false)
                        , m_callingPendingFunctors(false)
                        , m_threadId(CurrentThread::tid())
                        , m_poller(Poller::newDeafaultPoller(this))
                        , m_wakeupFd(createEventFd())
                        , m_wakeupChannel(new Channel(this, m_wakeupFd))
{
    LOG_DEBUG("EventLoop created %p in thread %d \n", this, m_threadId);
    if (t_loopInThisThread) {
        LOG_FATAL("Another EventLoop %p exists in this thread %d \n", t_loopInThisThread, m_threadId);
    } else {
        t_loopInThisThread = this;
    }

    // 设置wakeupfd的事件类型以及发生事件后的回调操作
    m_wakeupChannel->setReadCallback(std::bind(&EventLoop::handleRead, this));
    // 每一个eventLoop都将监听wakeupchannel的EPOLLIN读事件
    m_wakeupChannel->enableReading();
}

EventLoop::~EventLoop()
{
    m_wakeupChannel->disableAll();
    m_wakeupChannel->remove();
    ::close(m_wakeupFd);
    t_loopInThisThread = nullptr;
}

// 开启事件循环
void EventLoop::loop()
{
    m_looping = true;
    m_quit = false;
    LOG_INFO("EventLoop %p start looping \n", this);

    while (!m_quit) {
        
        m_activeChannels.clear();
        // 监听两类fd 一种是client的fd 一种是wakeupfd   m_activeChannels是存放的所有发生事件的Channel
        m_pollReturnTime = m_poller->poll(kPollTimeMs, &m_activeChannels);

        for (Channel* channel : m_activeChannels) {
            // Poller监听哪些Channel发生事件，然后上报给EventLoop，通过Channel处理相应的事件
            channel->handleEvent(m_pollReturnTime);
        }
        // 执行当前EventLoop事件循环需要处理的回调操作
        /** 
         * mainLoop事先注册一个回调cb(需要subloop来执行) wakeup subloop之后，执行下面的方法，执行之前mainLoop注册的cb操作 
         **/
        // 可能是mainLoop派发的回调操作，给subloop分配新的Channnel
        doPendingFunctors();
    }
    LOG_INFO("EventLoop %p stop looping...... \n", this);
    m_looping = false;
}

// 退出事件循环  1.loop在自己的线程中调用quit  2.在非loop的线程中，调用loop的quit
void EventLoop::quit()
{
    m_quit = true;

    // 如果是在其它线程中，调用的quit   在一个subloop(woker)中，调用了mainLoop(IO)的quit
    // 调用loop的quit函数时，不一定在自己的线程里面quit自己
    if (!isInLoopThread())  
    {
        wakeup();
    }
}

// 在当前loop中执行cb
void EventLoop::runInLoop(Functor cb)
{
    // 在当前loop对应的线程中执行回调cb
    if (isInLoopThread()) {
        cb();
    } else {
        // 在非当前loop对应的线程中执行回调cb，需要唤醒loop所在的线程执行回调cb
        queueInLoop(cb);
    }
}

// 把cb放入到容器中，唤醒loop所在的线程，执行cb
void EventLoop::queueInLoop(Functor cb)
{
    {
        std::unique_lock<std::mutex> lock(m_mutex);
        m_pendingFunctors.emplace_back(cb);
    }

    // 唤醒相应的，需要执行上面回调操作的loop线程
    // 添加m_callingPendingFunctors是当前loop正在执行回调，但是loop又有了新的回调，防止执行完一轮回调之后被阻塞，导致不会执行新添加的回调
    if (!isInLoopThread() || m_callingPendingFunctors) {
        // 唤醒loop所在的线程
        wakeup();
    }
}

// 向wakeupFd写数据，用来唤醒loop所在的线程
void EventLoop::wakeup()
{
    uint64_t one = 1;
    ssize_t n = write(m_wakeupFd, &one, sizeof(one));
    if (n != sizeof(one)) {
        LOG_ERROR("EventLoop::wakeup() writes %lu bytes instead of 8", n);
    }
}

// Channel通过EventLoop的方法来操作Poller
void EventLoop::updateChannel(Channel *channel)
{
    m_poller->updateChannel(channel);
}

void EventLoop::removeChannel(Channel *channel)
{
    m_poller->removeChannel(channel);
}

bool EventLoop::hasChannel(Channel *channel)
{
    return m_poller->hasChannel(channel);
}

// 唤醒对应线程的loop
void EventLoop::handleRead()
{
    uint64_t one = 1;
    ssize_t n = read(m_wakeupFd, &one, sizeof(one));
    if (n != sizeof(one)) {
        LOG_ERROR("EventLoop::handleRead() reads %lu bytes instead of 8", n);
    }
}

// 执行mainloop分配连接到当前子loop的回调
void EventLoop::doPendingFunctors()
{
    std::vector<Functor> functors;
    m_callingPendingFunctors = true;

    // 防止主反应堆在添加Channel时，被阻塞，影响程序效率
    {
        std::unique_lock<std::mutex> lock(m_mutex);
        functors.swap(m_pendingFunctors);
    }

    for (const auto& functor : functors) {
        // 执行当前loop需要执行的回调操作
        functor();
    }

    m_callingPendingFunctors = false;
}
